package top.sxlai.util.hbase;

import top.sxlai.entity.OpFile;
import top.sxlai.entity.OpFolder;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.NullComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author : sxl
 * @date : 2024/4/4 11:54
 * @Version: 1.0
 * 用于对表格数据的修改
 */
public class HbaseDML {
    //添加静态属性connection指向单例连接
    public static Connection connection = HbaseConnection.connection;

    /**
     * put方法用于向表格中添加数据
     *
     * @param namespace    命名空间
     * @param tableName    表格名称
     * @param rowKey       行键
     * @param columnFamily 列族
     * @param columnValues 列值
     * @throws IOException 异常
     */
    public static boolean putCell(String namespace, String tableName, String rowKey, String columnFamily, Map<String, String> columnValues) throws IOException {
        boolean flag = true;
        // 1.获取table
        Table table =
                connection.getTable(TableName.valueOf(namespace, tableName));
        // 2.调用相关方法插入数据
        //2.1创建put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        //2.2添加单元格数据
        //填入utf-8，消除中文乱码
        columnValues.entrySet()
                .forEach(entry -> put.addColumn(
                        Bytes.toBytes(columnFamily),
                        entry.getKey().getBytes(StandardCharsets.UTF_8),
                        entry.getValue().getBytes(StandardCharsets.UTF_8)));

        //2.3将对象写入1对应的方法
        try {
            table.put(put);
        } catch (IOException e) {
            flag = false;
            throw new RuntimeException(e);
        }
        // 3.关闭table
        table.close();
        return flag;
    }

    public static OpFolder getFolderCells(String namespace, String tableName, String rowKey, String columnFamily, List<String> columnNames) throws IOException {

        // 获取 table
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        //创建folder对象
        OpFolder opFolder = new OpFolder();
        try {
            // 创建 get 对象
            Get get = new Get(Bytes.toBytes(rowKey));
            // 添加列名
            for (String cn : columnNames) {
                get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(cn));
            }
            // 设置读取数据的版本
            get.readAllVersions();
            // 读取数据 得到 result 对象
            Result result = table.get(get);
            // 处理数据
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                // 这里可以根据需要处理数据，例如打印到控制台或者返回给调用者
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                byte[] value = CellUtil.cloneValue(cell);
                switch (qualifier) {
                    case "folderId":
                        opFolder.setFolderId(Bytes.toString(value));
                        break;
                    case "userId":
                        opFolder.setUserId(Bytes.toString(value));
                        break;
                    case "folderName":
                        opFolder.setFolderName(Bytes.toString(value));
                        break;
                    case "parentId":
                        opFolder.setParentId(Bytes.toString(value));
                        break;
                    case "hdfsPath":
                        opFolder.setHdfsPath(Bytes.toString(value));
                        break;
                    case "owner":
                        opFolder.setOwner(Bytes.toString(value));
                        break;
                    case "createTime":
                        opFolder.setCreateTime(Bytes.toString(value));
                        break;
                    case "status":
                        opFolder.setStatus(Bytes.toString(value));
                        break;
                    default:
                        // handle unknown qualifier
                        break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭 table
            table.close();
        }
        return opFolder;
    }

    /**
     * 扫描数据，返回分页查询
     *
     * @param namespace      命令空间
     * @param tableName      表格名称
     * @param rowKey         用户名作为行键
     * @param columnFamily   列族
     * @param columnNames    列名
     * @param selectQueryWay 查询方式
     * @param selectObj      选择对象
     * @param <T>            OpFile OpFolder
     * @return OpFile OpFolder
     * @throws IOException 异常
     */
    public static <T> List<T> filterScan(String namespace, String tableName, String rowKey,
                                         String columnFamily, List<String> columnNames, String selectQueryWay, String selectObj) throws IOException {
        List<T> resultList = new ArrayList<>();
        Connection connection = null;
        Table table = null;
        ResultScanner scanner = null;
        try {
            // 创建连接和表对象
            connection = ConnectionFactory.createConnection();
            table = connection.getTable(TableName.valueOf(namespace, tableName));
            // 设置 Scan 对象的属性
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(rowKey));
            scan.withStopRow(Bytes.toBytes(rowKey + Character.MAX_VALUE));
            // 添加过滤对象
            FilterList filterList = new FilterList();
            for (String columnName : columnNames) {
                SingleColumnValueFilter filter = new SingleColumnValueFilter(
                        Bytes.toBytes(columnFamily),
                        Bytes.toBytes(columnName),
                        CompareFilter.CompareOp.NOT_EQUAL,
                        new NullComparator());
                filter.setFilterIfMissing(true);
                filterList.addFilter(filter);
            }
            scan.setFilter(filterList);
            // 执行扫描
            scanner = table.getScanner(scan);
            // 处理扫描结果
//            int startRow = (pageNumber - 1) * pageSize;
//            int endRow = startRow + pageSize;
//            int count = 0;
            for (Result result : scanner) {
//                if (count >= startRow && count < endRow) {
                T obj;
                if ("file".equals(selectObj)) {
                    obj = parseFileResult(result, rowKey, selectQueryWay);
                } else {
                    obj = parseFolderResult(result, rowKey, selectQueryWay);
                }
                resultList.add(obj);
//                }
//                count++;
            }
        } finally {
            // 关闭资源
            if (scanner != null) {
                scanner.close();
            }
            if (table != null) {
                table.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
        return resultList;
    }

    private static <T> T parseFileResult(Result result, String rowKey, String selectQueryWay) {
        // 这里根据实际情况解析结果并封装成对象
        Cell[] cells = result.rawCells();
        OpFile opFile = new OpFile();
        //加个判断，看是要查询直接子级文件还是全部文件为 1 查询子级文件，为2查询全部文件
        if ("1".equals(selectQueryWay)) {
            for (Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                byte[] value = CellUtil.cloneValue(cell);
                String currentRowKey = new String(CellUtil.cloneRow(cell));
                if (isDirectChild2(currentRowKey, rowKey)) {
                    switch (qualifier) {
                        case "fileId":
                            opFile.setFileId(Bytes.toString(value));
                            break;
                        case "fileName":
                            opFile.setFileName(Bytes.toString(value));
                            break;
                        case "folderId":
                            opFile.setFolderId(Bytes.toString(value));
                            break;
                        case "typeId":
                            opFile.setTypeId(Bytes.toString(value));
                            break;
                        case "userId":
                            opFile.setUserId(Bytes.toString(value));
                            break;
                        case "createTime":
                            opFile.setCreateTime(Bytes.toString(value));
                            break;
                        case "owner":
                            opFile.setOwner(Bytes.toString(value));
                            break;
                        case "status":
                            opFile.setStatus(Bytes.toString(value));
                            break;
                        case "hdfsPath":
                            opFile.setHdfsPath(Bytes.toString(value));
                            break;
                        case "fileSize":
                            opFile.setFileSize(Bytes.toString(value));
                            break;
                        case "imageUrl":
                            opFile.setImageUrl(Bytes.toString(value));
                        default:
                            // handle unknown qualifier
                            break;
                    }
                }
            }
        } else  {
            for (Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                byte[] value = CellUtil.cloneValue(cell);
                switch (qualifier) {
                    case "fileId":
                        opFile.setFileId(Bytes.toString(value));
                        break;
                    case "fileName":
                        opFile.setFileName(Bytes.toString(value));
                        break;
                    case "folderId":
                        opFile.setFolderId(Bytes.toString(value));
                        break;
                    case "typeId":
                        opFile.setTypeId(Bytes.toString(value));
                        break;
                    case "userId":
                        opFile.setUserId(Bytes.toString(value));
                        break;
                    case "createTime":
                        opFile.setCreateTime(Bytes.toString(value));
                        break;
                    case "owner":
                        opFile.setOwner(Bytes.toString(value));
                        break;
                    case "status":
                        opFile.setStatus(Bytes.toString(value));
                        break;
                    case "hdfsPath":
                        opFile.setHdfsPath(Bytes.toString(value));
                        break;
                    case "fileSize":
                        opFile.setFileSize(Bytes.toString(value));
                        break;
                    case "imageUrl":
                        opFile.setImageUrl(Bytes.toString(value));
                    default:
                        // handle unknown qualifier
                        break;
                }
            }
        }
        return opFile.getHdfsPath() != null ? (T) opFile : null;
    }

    private static <T> T parseFolderResult(Result result, String rowKey, String selectQueryWay) {
        // 这里根据实际情况解析结果并封装成对象
        Cell[] cells = result.rawCells();
        // 创建一个新的 OpFolder 对象
        OpFolder opFolder = new OpFolder();
        // 根据 Cell 数组中的数据设置 OpFolder 对象的属性
        if ("2".equals(selectQueryWay)) {
            for (Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                byte[] value = CellUtil.cloneValue(cell);
                switch (qualifier) {
                    case "folderId":
                        opFolder.setFolderId(Bytes.toString(value));
                        break;
                    case "userId":
                        opFolder.setUserId(Bytes.toString(value));
                        break;
                    case "folderName":
                        opFolder.setFolderName(Bytes.toString(value));
                        break;
                    case "parentId":
                        opFolder.setParentId(Bytes.toString(value));
                        break;
                    case "hdfsPath":
                        opFolder.setHdfsPath(Bytes.toString(value));
                        break;
                    case "owner":
                        opFolder.setOwner(Bytes.toString(value));
                        break;
                    case "createTime":
                        opFolder.setCreateTime(Bytes.toString(value));
                        break;
                    case "status":
                        opFolder.setStatus(Bytes.toString(value));
                        break;
                    default:
                        // handle unknown qualifier
                        break;
                }

            }
        } else {
            for (Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                byte[] value = CellUtil.cloneValue(cell);
                // 获取当前行的行键
                String currentRowKey = new String(CellUtil.cloneRow(cell));
                if (isDirectChild(currentRowKey, rowKey)) {
                    switch (qualifier) {
                        case "folderId":
                            opFolder.setFolderId(Bytes.toString(value));
                            break;
                        case "userId":
                            opFolder.setUserId(Bytes.toString(value));
                            break;
                        case "folderName":
                            opFolder.setFolderName(Bytes.toString(value));
                            break;
                        case "parentId":
                            opFolder.setParentId(Bytes.toString(value));
                            break;
                        case "hdfsPath":
                            opFolder.setHdfsPath(Bytes.toString(value));
                            break;
                        case "owner":
                            opFolder.setOwner(Bytes.toString(value));
                            break;
                        case "createTime":
                            opFolder.setCreateTime(Bytes.toString(value));
                            break;
                        case "status":
                            opFolder.setStatus(Bytes.toString(value));
                            break;
                        default:
                            // handle unknown qualifier
                            break;
                    }
                }
            }
        }
        // 返回当有路径了的 OpFolder 对象
        return opFolder.getHdfsPath() != null ? (T) opFolder : null;
    }

    private static boolean isDirectChild(String currentRowKey, String givenRowKey) {
        // 检查路径是否以文件后缀名结尾
        if (currentRowKey.contains(".")) {
            return false;
        }
        // 将行键分割成路径数组
        String[] currentPath = currentRowKey.split("/");
        String[] givenPath = givenRowKey.split("/");
        // 如果当前路径比给定路径短1，且以给定路径为前缀，则是直接子级
        return currentPath.length == givenPath.length + 1 && currentRowKey.startsWith(givenRowKey);
    }

    private static boolean isDirectChild2(String currentRowKey, String givenRowKey) {
        // 检查路径是否以文件后缀名结尾
        if (!currentRowKey.contains(".")) {
            return false;
        }
        // 将行键分割成路径数组
        String[] currentPath = currentRowKey.split("/");
        String[] givenPath = givenRowKey.split("/");
        // 如果当前路径比给定路径短1，且以给定路径为前缀，则是直接子级
        return currentPath.length == givenPath.length + 1 && currentRowKey.startsWith(givenRowKey);
    }

    /**
     * 删除HBase表中的数据
     *
     * @param nameSpace
     * @param tableName
     * @param rowKey
     * @param family
     * @param columnNames
     * @throws IOException
     */
    public static void deleteColumns(String nameSpace, String tableName,
                                     String rowKey, String family, List<String> columnNames) throws IOException {
        try (Table table = connection.getTable(TableName.valueOf(nameSpace, tableName))) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            for (String columnName : columnNames) {
                delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName));
            }
            table.delete(delete);
        } catch (IOException e) {
            // 处理异常或者抛出更高层次的异常
            e.printStackTrace();
        }
    }

    /**
     * 更新HBase表中的数据
     *
     * @param nameSpace
     * @param tableName
     * @param rowKey
     * @param family
     * @param columnName
     * @param newValue
     * @throws IOException
     */
    public static void updateColumn(String nameSpace, String tableName,
                                    String rowKey, String family, String columnName, String newValue) throws IOException {
        try (Table table = connection.getTable(TableName.valueOf(nameSpace, tableName))) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName), Bytes.toBytes(newValue));
            table.put(put);
        } catch (IOException e) {
            // 处理异常或者抛出更高层次的异常
            e.printStackTrace();
        }
    }

    /**
     * 修改行键
     *
     * @param nameSpace
     * @param tableName
     * @param family
     * @param oldRowKey
     * @param newRowKey
     * @throws IOException
     */
    public static boolean updateFileRowKey(String nameSpace, String tableName
            , String oldRowKey, String family, String newRowKey) throws IOException {
        boolean flag = true;
        try (Table table = connection.getTable(TableName.valueOf(nameSpace, tableName))) {
            // 1. 获取要修改的行的数据
            Get get = new Get(Bytes.toBytes(oldRowKey));
            get.addFamily(Bytes.toBytes(family));
            Result result = table.get(get);
            if (!result.isEmpty()) {
                // 2. 使用新的行键创建一个新的 Put 对象
                Put put = new Put(Bytes.toBytes(newRowKey));
                // 3. 将原始行的数据复制到新的 Put 对象中
                for (Cell cell : result.rawCells()) {
                    put.addColumn(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell), CellUtil.cloneValue(cell));
                }
                // 4. 在新的行上插入新的数据
                table.put(put);
                // 5. 删除原始行
                Delete delete = new Delete(Bytes.toBytes(oldRowKey));
                table.delete(delete);
            } else {
                System.out.println("未找到要修改的行");
                flag = false;
            }
        } catch (IOException e) {
            // 处理异常或者抛出更高层次的异常
            e.printStackTrace();
        }
        return flag;
    }

    /**
     * 获取文件夹下的所有行键
     *
     * @param namespace
     * @param fileTableName
     * @param hdfsPath      行键
     * @return
     */
    public static List<String> getAllKeys(String namespace, String fileTableName, String hdfsPath) {
        List<String> fileRowKeys = new ArrayList<>();
        // 获取HBase连接
        try {
            // 获取表对象
            TableName tableName = TableName.valueOf(namespace, fileTableName);
            org.apache.hadoop.hbase.client.Table table = connection.getTable(tableName);
            // 构建扫描器
            Scan scan = new Scan();
            // 获取结果扫描器
            try (ResultScanner scanner = table.getScanner(scan)) {
                // 遍历结果并提取行键
                for (Result result : scanner) {
                    byte[] rowKeyBytes = result.getRow();
                    if (rowKeyBytes != null) {
                        String rowKey = Bytes.toString(rowKeyBytes);
                        // 如果行键以指定路径开头，则添加到列表中
                        if (rowKey.startsWith(hdfsPath)) {
                            fileRowKeys.add(rowKey);
                        }
                    }
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return fileRowKeys;
    }
}
